Skip to content

[SPARK-58093][SQL] Add ASOF JOIN SQL syntax with MATCH_CONDITION#57194

Open
srielau wants to merge 23 commits into
apache:masterfrom
srielau:SPARK-58093-ASOF-JOIN
Open

[SPARK-58093][SQL] Add ASOF JOIN SQL syntax with MATCH_CONDITION#57194
srielau wants to merge 23 commits into
apache:masterfrom
srielau:SPARK-58093-ASOF-JOIN

Conversation

@srielau

@srielau srielau commented Jul 11, 2026

Copy link
Copy Markdown
Contributor

Wire the SQL Reference Spec ASOF JOIN surface into the existing AsOfJoin logical operator, reusing the DataFrame execution path (RewriteAsOfJoin / SortMergeAsOfJoinExec). Adds parser grammar, analysis resolution for USING and MATCH_CONDITION operands, spec error conditions, and parser/SQL integration tests.

What changes were proposed in this pull request?

This PR adds SQL syntax for ASOF JOIN with a required MATCH_CONDITION clause and optional ON / USING, per the SQL Language Reference spec (Snowflake / Calcite MATCH_CONDITION family).

Syntax supported (v1):

[ INNER | LEFT [ OUTER ] ] ASOF JOIN right_table
  MATCH_CONDITION ( left_expr comparison_operator right_expr )
  [ ON boolean_expression | USING ( column_name [, ...] ) ]

where comparison_operator is one of >=, >, <=, <.

Example:

SELECT t.trade_time, t.symbol, q.bid_price
FROM trades t
ASOF JOIN quotes q
  MATCH_CONDITION (t.trade_time >= q.quote_time)
  ON t.symbol = q.symbol;

Implementation layers (new vs reused):

Layer Change
Parser ASOF, MATCH_CONDITION keywords; asofJoinType, asofJoinCriteria, matchComparison grammar rules
AstBuilder withAsOfJoin() builds AsOfJoin.fromMatchCondition(...)
Analysis New ResolveAsOfJoin rule: expands USING via NaturalAndUsingJoinResolution, normalizes MATCH_CONDITION operands into asOfCondition + orderExpression, validates operand expressions/types
Logical plan Extends AsOfJoin with deferred matchComparison / usingColumns fields; adds MatchComparisonOperator and AsOfJoin.fromMatchCondition
Execution SQL ASOF JOIN requires spark.sql.join.sortMergeAsOfJoin.enabled=true (sort-merge only; RewriteAsOfJoin / min_by path is not used for SQL). Extends SortMergeAsOfJoinExec for multi-column sort keys and complex MATCH_CONDITION types
Errors Five error classes: ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION, ..._INVALID_OPERATOR, ..._INVALID_TYPE, ..._TABLE_REFERENCE, plus AS_OF_JOIN.SORT_MERGE_REQUIRED when sort-merge is disabled

Explicitly out of scope (deferred per spec): TOLERANCE, NEAREST direction, RIGHT / FULL / CROSS / SEMI / ANTI ASOF forms.

Related prior art in Spark: DataFrame.joinAsOf (PySpark/Scala), AsOfJoin logical node, SortMergeAsOfJoinSuite, DataFrameAsOfJoinSuite.

Why are the changes needed?

ASOF (as-of / closest-match) join is a standard temporal/analytics pattern — attaching the nearest preceding quote to each trade, finding the next maintenance window for an alert, etc. Spark already implements the semantics via the DataFrame API, but SQL users must hand-write LEFT JOIN + ROW_NUMBER + QUALIFY rewrites. Native SQL syntax lowers the barrier and aligns with industry dialects (Snowflake, Calcite/Pinot/Doris MATCH_CONDITION family).

Does this PR introduce any user-facing change?

Yes.

Before: ASOF JOIN was not valid SQL; closest-match joins were only available through DataFrame.joinAsOf.

After: SQL queries may use ASOF JOIN / LEFT ASOF JOIN with MATCH_CONDITION. Behavior matches the spec: asymmetric left-driven lookup, at most one right row per left row, INNER drops unmatched left rows, LEFT preserves them with NULL right-side columns.

Operand order inside MATCH_CONDITION is immaterial (t.ts >= q.tsq.ts <= t.ts); the analyzer normalizes to a canonical left-table-on-left form before planning.

How was this patch tested?

Added unit and integration tests; full compile/test suite to run on CI (local build blocked by missing netty 4.2.16 artifacts on the dev Maven proxy after rebasing onto latest master).

Parser tests (PlanParserSuite):

  • INNER / LEFT ASOF JOIN with MATCH_CONDITION and ON
  • USING clause
  • Swapped operand order (u.a <= t.a)
  • Rejection of = in MATCH_CONDITION (parse error)

SQL integration tests (AsOfJoinSQLSuite):

  • Spec examples: trades/quotes with ON, LEFT ASOF preserving unmatched rows, USING equivalence, first-following match with <=

Existing coverage reused (unchanged): DataFrameAsOfJoinSuite, SortMergeAsOfJoinSuite, RewriteAsOfJoinSuite exercise the shared execution path.

Suggested CI commands for reviewers:

build/sbt "sql/catalyst/testOnly org.apache.spark.sql.catalyst.parser.PlanParserSuite -- -z \"asof join\""
build/sbt "sql/testOnly org.apache.spark.sql.AsOfJoinSQLSuite"
build/sbt "sql/testOnly org.apache.spark.sql.DataFrameAsOfJoinSuite"

Was this patch authored or co-authored using generative AI tooling?

Yes. Generated-by: Cursor (Claude agent), with human review and editing by the PR author.

Wire the SQL Reference Spec ASOF JOIN surface into the existing AsOfJoin
logical operator, reusing the DataFrame execution path (RewriteAsOfJoin /
SortMergeAsOfJoinExec). Adds parser grammar, analysis resolution for USING
and match operands, spec error conditions, and parser/SQL integration tests.
@srielau srielau marked this pull request as draft July 11, 2026 05:05
srielau added 12 commits July 11, 2026 05:47
…pile errors

Update case-class pattern matches for the two new AsOfJoin fields, fix
conf.resolver usage, replace expr.preOrder with expr.collect, and drop a
duplicate RowOrdering import.
… arity

The error helper only accepts the two MATCH_CONDITION operands; drop the
unused left/right AttributeSet arguments from the throw site.
Rule.ruleId is a lazy val; overriding it with a non-lazy val fails
compilation. The base Rule implementation already derives the same id
from the class name, so drop the override entirely.
Wrap long lines and reorder expression imports so the braced import
precedes expressions.aggregate.
Store parsed MATCH_CONDITION operands in AsOfMatchCondition so mapExpressions
does not corrupt the tuple shape. Resolve operands in ResolveReferences and
materialize asOfCondition in ResolveAsOfJoin, including USING column projection.
…enabled

Default the new SQL syntax off while under development. AstBuilder rejects
ASOF JOIN at parse time when disabled; tests opt in via SQLConf.
…iance doc

SQLKeywordSuite requires new lexer keywords to be listed in
sql-ref-ansi-compliance.md with correct reserved/non-reserved status.
ASOF is a Spark extension, not SQL standard, so it belongs in
ansiNonReserved rather than being reserved under ANSI mode.
…sing

Restrict matchComparison to valueExpression so boolean AND is rejected at
parse time. Tighten type validation for incompatible operand pairs and
prepare struct/tuple ordering for field-wise distance expressions.
…expressions

Expose parsed MATCH_CONDITION operand trees to analysis pruning and
expression rewriting so functions, interval arithmetic, and ON predicates
all resolve before ResolveAsOfJoin materializes the join condition.
…erands

Recursively build min_by order expressions for nested structs (leaf
flattening) and arrays (ZipWith over element-wise diffs). Use signed
comparison distance for orderable non-numeric types like STRING. Add
AsOfJoinSQLSuite coverage for nested struct, ARRAY<INT>, and struct tuple
operands on the min_by rewrite path.
@srielau srielau marked this pull request as ready for review July 12, 2026 04:04
srielau added 7 commits July 12, 2026 04:08
Place TreePattern before TreePattern._ and keep DataTypeUtils imports
adjacent per scalastyle rules.
Regenerate keywords golden files to include ASOF and MATCH_CONDITION
from the lexer vocabulary. Update bitwise parse-error goldens after the
ASOF JOIN grammar change shifts the reported token for `> >>` syntax.
Add ASOF and MATCH_CONDITION to the SPARK-43119 hardcoded keyword list
returned by GetInfo(CLI_ODBC_KEYWORDS).
Reorder ASOF_JOIN_MATCH_CONDITION_* entries before ASSIGNMENT_ARITY_MISMATCH
and AS_OF_JOIN in error-conditions.json to satisfy SparkThrowableSuite
key ordering (ORDER_MAP_ENTRIES_BY_KEYS).
Add ASOF and MATCH_CONDITION to the hardcoded getSQLKeywords expectation
after the new ASOF JOIN grammar keywords were registered.
Parse MATCH_CONDITION comparisons as booleanExpression instead of a
dedicated matchComparison rule, which disturbed unrelated expression
error reporting (e.g. CHECK constraints). Move ASOF JOIN to a trailing
grammar alternative and validate allowed comparison operators in
AstBuilder. Update the join_ parse-error golden after the new ASOF
alternative changes ANTLR error recovery.
Revert the reported error token for `SELECT 20181117 > >> 2` back to
'>>' after removing the ASOF matchComparison grammar rule restored the
original ANTLR error recovery behavior.
@srielau

srielau commented Jul 13, 2026

Copy link
Copy Markdown
Contributor Author

@sarutak Are you still actively working on the sort-merge-as-join feature? I need some extensions to generalize the feature. It woudl be helpful to know if you are having anything in flight yourself, or I can extend upon it.

SQL ASOF JOIN now requires spark.sql.join.sortMergeAsOfJoin.enabled.
Extend SortMergeAsOfJoinExec for multi-column MATCH_CONDITION sort keys,
fix LEFT OUTER null right-side projection, and add AsOfJoinSortMergeSQLSuite
covering scalar types, struct/array operands, and forward match semantics.

@srielau srielau left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code review summary

Overall this is a well-structured PR: parser → AsOfJoin.fromMatchConditionResolveAsOfJoinSortMergeAsOfJoinExec is coherent, feature gating is appropriate, and positive-path coverage in AsOfJoinSortMergeSQLSuite is strong.

No P0 correctness bugs found. A few items below are worth addressing before merge (inline comments have details).

Before merge (recommended):

  1. Fix array element type check (sameType vs ==) in buildOrderExpression
  2. Add analysis negative tests for ASOF_JOIN_MATCH_CONDITION_* error classes
  3. Add a LEFT OUTER regression test for the nullRightRow / NOT NULL catalog column case
  4. Update the PR description — latest commit requires sort-merge for SQL and extends SortMergeAsOfJoinExec, which contradicts the current "Execution: No changes" bullet

Nice-to-have (can follow up):

  • Throw ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR for compound MATCH_CONDITION instead of generic PARSE_SYNTAX_ERROR
  • Strict > / < and forward-array execution tests
  • sql-ref syntax documentation and dual-config callout

rightOperand: Expression,
operator: MatchComparisonOperator): Expression = {
(leftOperand.dataType, rightOperand.dataType) match {
case (ArrayType(leftElem, _), ArrayType(rightElem, _)) if leftElem == rightElem =>

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1: Use structural type equality for array elements

This gate uses reference equality (leftElem == rightElem), while matchSortExpressions (line 2704) uses leftStruct.sameType(rightStruct) and AsOfJoinValidation.areStructurallyComparableTypes allows structurally comparable ARRAY<STRUCT<...>> operands with different field names.

For such operands, analysis accepts the query but buildOrderExpression falls through to buildLeafOrderExpression on the whole array, which is incorrect for forward (<=/<) joins that rely on element-wise distance.

Suggested fix:

case (ArrayType(leftElem, _), ArrayType(rightElem, _))
    if DataTypeUtils.sameType(leftElem, rightElem) =>

|""".stripMargin)
}

test("SQL ASOF JOIN requires sort-merge conf") {

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1: Add analysis negative tests

AsOfJoinValidation rejects subqueries, aggregates, window functions, non-deterministic expressions, incompatible types, and cross-side operands — but none of these paths are covered by SQL integration tests today (only parser-level = rejection in AsOfJoinSQLSuite).

Consider adding checkError tests for at least:

  • ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE (e.g. MATCH_CONDITION (t.a + r.b >= r.c))
  • ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION (e.g. subquery or rand() in operand)
  • ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE (e.g. INT vs STRING)

private val nullRightRow = new GenericInternalRow(rightOutput.length)
// Materialize an all-null right row as UnsafeRow. GenericInternalRow cannot be
// passed through identity UnsafeProjection when right columns are NOT NULL.
private val nullRightRow: InternalRow = {

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1: Regression test for nullRightRow fix

This fix addresses a real bug: GenericInternalRow null padding fails UnsafeProjection when right-side catalog columns are NOT NULL. The LEFT OUTER tests use VALUES temp views without NOT NULL constraints, so this path isn't exercised in CI.

Consider a test with an explicit NOT NULL right-side schema (catalog table or CREATE TABLE ... NOT NULL) where an unmatched left row must produce null right columns.

case GreaterThan(left, right) => (left, GreaterThanOp, right)
case LessThanOrEqual(left, right) => (left, LessThanOrEqualOp, right)
case LessThan(left, right) => (left, LessThanOp, right)
case _ =>

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: Dead error class / misleading parse error for compound MATCH_CONDITION

MATCH_CONDITION (t.a >= u.a AND t.b >= u.b) hits this branch and surfaces PARSE_SYNTAX_ERROR with error: ')'. Meanwhile ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR is defined in error-conditions.json but never thrown.

Consider rejecting non-singular comparisons here with ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR (or a clearer dedicated message).

Row(Date.valueOf("2026-06-29"), Date.valueOf("2026-06-29")) :: Nil)
}

test("INT scalar MATCH_CONDITION") {

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: Test coverage gaps

Execution tests cover >= and <= well but not strict > / < (different tie-breaking semantics). Array tests use backward >= only; forward array joins use distance-based early termination in findBestForwardNearest, which is the more fragile path.

Low priority, but one backward > and one forward <= on ARRAY<INT> would close the gap.

@sarutak

sarutak commented Jul 13, 2026

Copy link
Copy Markdown
Member

@srielau

@sarutak Are you still actively working on the sort-merge-as-join feature? I need some extensions to generalize the feature. It woudl be helpful to know if you are having anything in flight yourself, or I can extend upon it.

I have been considering adding syntax for AS-OF JOIN, but I haven't started working on it yet. Please go ahead and take the lead on this.

srielau added 2 commits July 13, 2026 18:20
Use DataTypeUtils.sameType for array MATCH_CONDITION operands, throw
ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR for invalid/compound operators,
and expand SQL integration tests for analysis errors, strict operators,
and NOT NULL null-padding.
Validate MATCH_CONDITION operands that reference both join inputs before
materializing the comparison, so invalid SQL like (t.col + r.col >= r.col)
raises ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE instead of an internal
unresolved-operator error.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants